package day03;


import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;

import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * @Description: 使用状态变量求平均值
 * @Author ZYX
 * @Date 2022/2/10 18:44
 * @Version 1.0
 */
public class Demo07 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        env.addSource(new SourceFunction<Integer>() {
            private boolean running = true;
            private Random random = new Random();

            @Override
            public void run(SourceContext<Integer> sourceContext) throws Exception {
                while (running){
                    sourceContext.collect(random.nextInt(20));
                    TimeUnit.MILLISECONDS.sleep(500L);
                }
            }

            @Override
            public void cancel() {
                running = false;
            }
        }).keyBy(obj->1)
                .process(new KeyedProcessFunction<Integer, Integer, Double>() {

                    private ListState<Integer> listState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        listState = getRuntimeContext().getListState(new ListStateDescriptor<Integer>("list列表", Types.INT));
                    }

                    @Override
                    public void processElement(Integer value, Context context, Collector<Double> out) throws Exception {
                        listState.add(value);
                        Integer sum = 0;
                        Integer count = 0;

                        for (Integer integer : listState.get()) {
                            sum += integer;
                            count += 1;
                        }

                        out.collect((double) (sum/count));
                    }
                }).print();


        env.execute();
    }

}
